#pragma once

#include "requestor.hpp"
#include "rpc_caller.hpp"
#include "rpc_registry.hpp"
#include "../common/dispatcher.hpp"
#include "rpc_topic.hpp"

namespace JsonMuduoRPC
{
    namespace client
    {
        class RegistryClient
        {
        public:
            using ptr = std::shared_ptr<RegistryClient>;

            RegistryClient(const std::string &ip, const uint16_t port)
                : _requestor(std::make_shared<Requestor>()),
                  _provider(std::make_shared<Provider>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())

            {
                // 设置回调
                auto cb = std::bind(&Requestor::onResponse, _requestor.get(),
                                    std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<ServiceResponse>(MType::RSP_SERVICE, cb);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallBack(std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                      std::placeholders::_1, std::placeholders::_2));
                _client->connect();
            }

            bool registryRequest(const std::string &method, const Address &host)
            {
                return _provider->registryRequest(_client->connection(), method, host);
            }

        private:
            BaseClient::ptr _client;
            Requestor::ptr _requestor;
            Provider::ptr _provider;
            Dispatcher::ptr _dispatcher;
        };

        class DiscoveryClient
        {
        public:
            using ptr = std::shared_ptr<DiscoveryClient>;

            DiscoveryClient(const std::string &ip, const uint16_t port, const Discover::OfflineCallBack &offline_callback)
                : _requestor(std::make_shared<Requestor>()),
                  _discover(std::make_shared<Discover>(_requestor, offline_callback)),
                  _dispatcher(std::make_shared<Dispatcher>())

            {
                // 设置回调
                auto rsp_cb = std::bind(&Requestor::onResponse, _requestor.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                auto req_cb = std::bind(&Discover::onServiceRequest, _discover.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<ServiceResponse>(MType::RSP_SERVICE, rsp_cb);
                _dispatcher->registerHandler<ServiceRequest>(MType::REQ_SERVICE, req_cb);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallBack(std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                      std::placeholders::_1, std::placeholders::_2));
                _client->connect();
            }

            bool serviceDiscovery(const std::string &method, Address &host)
            {
                return _discover->serviceDiscovery(_client->connection(), method, host);
            }

        private:
            BaseClient::ptr _client;
            Requestor::ptr _requestor;
            Discover::ptr _discover;
            Dispatcher::ptr _dispatcher;
        };

        class RPCClient
        {
        public:
            using ptr = std::shared_ptr<RPCClient>;

            // enableDiscovery--是否启用服务发现功能，也决定了传入的地址信息是注册中心的地址，还是服务提供者的地址
            // 不启用——传入的地址信息是服务提供者的地址
            // 启用——传入的地址信息是注册中心的地址
            RPCClient(bool enableDiscovery, const std::string &ip, const uint16_t port)
                : _enableDiscovery(enableDiscovery),
                  _requestor(std::make_shared<Requestor>()),
                  _caller(std::make_shared<RPCCaller>(_requestor)),
                  _dispatcher(std::make_shared<Dispatcher>())

            {
                // 设置回调
                auto rsp_cb = std::bind(&Requestor::onResponse, _requestor.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<RPCResponse>(MType::RSP_RPC, rsp_cb);

                // 如果服务发现开启
                if (_enableDiscovery)
                {
                    auto offline_cb = std::bind(&RPCClient::deleteClient, this, std::placeholders::_1);
                    _discovery_client = std::make_shared<DiscoveryClient>(ip, port, offline_cb);
                }
                else // 如果服务发现未开启，直接实例化一个客户端client
                {
                    _client = ClientFactory::create(ip, port);
                    _client->setMessageCallBack(std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                          std::placeholders::_1, std::placeholders::_2));
                    _client->connect();
                }
            }

            // 以同步的方式发送请求，等收到响应结果后返回
            bool call(const std::string &method, const Json::Value &params, Json::Value &result)
            {
                BaseClient::ptr client = getClient(method);
                if (client.get() != nullptr)
                    return _caller->call(client->connection(), method, params, result);
                else
                    return false;
            }

            // 以异步的方式发送请求，发起调用后立即返回，在想获取结果的时候进行获取
            // 通过设置回调函数，在回调函数中设置promise的值（结果）
            bool call(const std::string &method, const Json::Value &params, RPCCaller::JsonAsyncResponse &result)
            {
                BaseClient::ptr client = getClient(method);
                if (client.get() != nullptr)
                    return _caller->call(client->connection(), method, params, result);
                else
                    return false;
            }

            bool call(const std::string &method, const Json::Value &params, const RPCCaller::JsonResponseCallBack &cb)
            {
                BaseClient::ptr client = getClient(method);
                if (client.get() != nullptr)
                    return _caller->call(client->connection(), method, params, cb);
                else
                    return false;
            }

        private:
            // 计算自定义类型Address的哈希值
            struct AddressHash
            {
                size_t operator()(const Address &host) const
                {
                    std::string str = host.first + std::to_string(host.second);
                    return std::hash<std::string>{}(str);
                }
            };

            BaseClient::ptr newClient(const Address &host)
            {
                BaseClient::ptr client = ClientFactory::create(host.first, host.second);
                client->setMessageCallBack(std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                     std::placeholders::_1, std::placeholders::_2));
                client->connect();

                putClient(host, client);

                return client;
            }

            BaseClient::ptr getClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _client_map.find(host);
                if (it == _client_map.end())
                    return BaseClient::ptr();

                return it->second;
            }

            BaseClient::ptr getClient(const std::string &method)
            {
                // 如果服务发现开启
                if (_enableDiscovery)
                {
                    // 先进行服务发现
                    Address host;
                    bool ret = _discovery_client->serviceDiscovery(method, host);
                    if (ret == false)
                    {
                        LOG(ERROR, "当前 %s 服务，没有找到服务提供者！", method.c_str());
                        return BaseClient::ptr();
                    }

                    // 找到提供服务的主机host对应的客户端连接，没有则创建
                    BaseClient::ptr client = getClient(host);
                    return client.get() == nullptr ? newClient(host) : client;
                }
                else
                    return _client;
            }

            void putClient(const Address &host, BaseClient::ptr &client)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _client_map.insert(std::make_pair(host, client));
            }

            void deleteClient(const Address &host)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _client_map.erase(host);
            }

        private:
            // 一个主机已经被哪个客户端发起过rpc请求，长连接思想
            // 客户端发起rpc请求并收到响应后不需要关闭连接，而是进入连接池
            // 当要发送同样的rpc请求时，就可以用连接池中的客户端进行发送
            std::unordered_map<Address, BaseClient::ptr, AddressHash> _client_map; // 用于开启服务发现功能
            BaseClient::ptr _client;                                               // 用于关闭服务发现功能
            DiscoveryClient::ptr _discovery_client;

            bool _enableDiscovery;
            std::mutex _mutex;
            Requestor::ptr _requestor;
            RPCCaller::ptr _caller;
            Dispatcher::ptr _dispatcher;
        };

        class TopicClient
        {
        public:
            using ptr = std::shared_ptr<TopicClient>;

            TopicClient(const std::string &ip, const uint16_t port)
                : _requestor(std::make_shared<Requestor>()),
                  _dispatcher(std::make_shared<Dispatcher>()),
                  _topic_manager(std::make_shared<TopicManager>(_requestor))

            {
                // 设置回调
                auto rsp_cb = std::bind(&Requestor::onResponse, _requestor.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                auto req_cb = std::bind(&TopicManager::onPublish, _topic_manager.get(),
                                        std::placeholders::_1, std::placeholders::_2);

                _dispatcher->registerHandler<TopicResponse>(MType::RSP_TOPIC, rsp_cb);
                _dispatcher->registerHandler<TopicRequest>(MType::REQ_TOPIC, req_cb);

                _client = ClientFactory::create(ip, port);
                _client->setMessageCallBack(std::bind(&Dispatcher::onMessage, _dispatcher.get(),
                                                      std::placeholders::_1, std::placeholders::_2));
                _client->connect();
            }

            bool create(const std::string& topic_name)
            {
                return _topic_manager->create(_client->connection(), topic_name);
            }

            bool remove(const std::string& topic_name)
            {
                return _topic_manager->remove(_client->connection(), topic_name);
            }

            bool subscribe(const std::string &topic_name, const TopicManager::SubscriveCallBack& cb)
            {
                return _topic_manager->subscribe(_client->connection(), topic_name, cb);
            }

            bool cancel(const std::string& topic_name)
            {
                return _topic_manager->cancel(_client->connection(), topic_name);
            }

            bool publish(const std::string &topic_name, const std::string &msg)
            {
                return _topic_manager->publish(_client->connection(), topic_name, msg);
            }

            void shutdown()
            {
                _client->shutdown();
            }

        private:
            BaseClient::ptr _client;
            Requestor::ptr _requestor;
            Dispatcher::ptr _dispatcher;
            TopicManager::ptr _topic_manager;
        };
    }
}